#ifndef __M_CCONECTION_H__
#define __M_CCONECTION_H__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include"muduo/net/EventLoopThread.h"
#include "muduo/net/TcpClient.h"

#include"../mqcommon/mq_msg.pb.h"
#include"../mqcommon/helper.hpp"
#include"../mqcommon/logger.hpp"
#include"../mqcommon/mq_proto.pb.h"
#include"../mqcommon/mq_threadpool.hpp"

#include"mqc_channel.hpp"
#include"mqc_worker.hpp"

#include<google/protobuf/map.h>

namespace mq{

    using namespace std::placeholders;
    class ClientConnection{
        public:
            ClientConnection(const std::string& ip,const int& port,const AsyncWorker::ptr& worker)
            :_worker(worker),
            _client(_worker->_eventLoop.startLoop(), muduo::net::InetAddress(ip,port), "Client"),
            _latch(1),
            _dispatcher(std::bind(&ClientConnection::onUnknownMessage, this, _1, _2, _3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, _1, _2, _3))),
            _channelManager(std::make_shared<ClientChannelManager>())
            {

                _dispatcher.registerMessageCallback<mq::basicConsumerResponse>(
                    std::bind(&ClientConnection::onbasicConsumer, this, _1, _2, _3));
                _dispatcher.registerMessageCallback<mq::basicCommonResponse>(
                    std::bind(&ClientConnection::onbasicCommon, this, _1, _2, _3));
                    
                _client.setConnectionCallback(
                    std::bind(&ClientConnection::onConnection, this, _1));
                _client.setMessageCallback(
                    std::bind(&ProtobufCodec::onMessage, _codec.get(), _1, _2, _3));
            }

            void connect()
            {
                _client.connect();
                _latch.wait();
            }

            ClientChannel::ptr openChannel(){
                ClientChannel::ptr channel = _channelManager->create(_conn,_codec);
                LOG_DEBUGM("_channelManager create done...");
                int ret = channel->openChannel();
                LOG_DEBUGM("channel openChannel done...");

                if(ret == false){
                    LOG_ERRORM("openChannel false...");
                    return ClientChannel::ptr();
                }
                return channel;
            }

            bool closeChannel(const ClientChannel::ptr& channel){
                ClientChannel::ptr cp = _channelManager->getChannel(channel->getCid());
                int ret = cp->closeChannel();
                if(ret == false){
                    LOG_ERRORM("closeChannel false...");
                    return false;
                }
                return true;
            }

        private:
            void onConnection(const muduo::net::TcpConnectionPtr& conn) {
                std::cout << "onConnection called, connected: " << conn->connected() 
                        << ", conn pointer: " << conn.get() << std::endl;
                if (conn->connected()) {
                    std::cout << "获取conn成功: " << conn.get() << std::endl;
                    _conn = conn;//有了
                    _latch.countDown();
                    std::cout << "_conn set to: " << _conn.get() << std::endl;
                } else {
                    std::cout << "连接断开" << std::endl;
                    _conn.reset();
                }
            }

            void onUnknownMessage(const muduo::net::TcpConnectionPtr&,
                                    const MessagePtr& message,
                                    muduo::Timestamp)
            {
                LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            }

            void onbasicConsumer(const muduo::net::TcpConnectionPtr& conn,
                                    const basicConsumerResponsePtr& resp,
                                    muduo::Timestamp)
            {
                ClientChannel::ptr channel = _channelManager->getChannel(resp->cid());
                if(channel.get() == nullptr){
                    LOG_ERRORM("ConsumeResp false...");
                    return ;
                }
                if(!resp){
                    LOG_ERRORM("basicConsumerResponse 为空...");
                    return ;
                }
                _worker->_pool.push([channel,resp](){
                    channel->consume(resp);
                });
            }

            void onbasicCommon(const muduo::net::TcpConnectionPtr& conn,
                                    const basicCommonResponsePtr& resp,
                                    muduo::Timestamp)
            {
                ClientChannel::ptr channel = _channelManager->getChannel(resp->cid());
                if(channel.get() == nullptr){
                    LOG_ERRORM("CommonResp false...");
                    return ;
                }
                channel->putBasicResponse(resp);
            }

        muduo::net::TcpConnectionPtr _conn;
        AsyncWorker::ptr _worker;
        muduo::net::TcpClient _client;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        muduo::CountDownLatch _latch;
        ClientChannelManager::ptr _channelManager;    
    };
    
}


#endif